package com.jltx.xq.coroutinesandrxpro.rxjava

import android.os.Handler
import android.util.Log
import io.reactivex.Observable
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.schedulers.Schedulers
import io.reactivex.subjects.PublishSubject
import java.util.concurrent.TimeUnit

/**
 * @author aidenquan
 *         Created by junlintianxia on 2021/08/09.
 */
object RxJava2Learn3 {

    /**
     * 依次发送数据到下游
     */
    fun customIntervalSub() {
        val disposabe =
            customInterval()
                .delay(2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({ t -> Log.d(RxJava2Learn1.TAG, "Rx customIntervalSub Next $t") }
                ) { t -> Log.d(RxJava2Learn1.TAG, "Rx customIntervalSub Error $t") }
    }

    private fun customInterval(): Observable<Int> = Observable.create { emitter ->
        for (i in 1..5) {
            if (!emitter.isDisposed) {
                emitter.onNext(i)
            } else {
                break
            }
        }
        emitter.onComplete()
    }

    /**
     * rxjava订阅者和被订阅者，类似于协程里面的广播通道
     */
    fun publishSubjectInterval() {
        val subject: PublishSubject<Int> = PublishSubject.create()
        subject.subscribe {
            Log.d(
                RxJava2Learn1.TAG,
                "Rx publishSubjectInterval consume $it ${Thread.currentThread()}"
            )
        }

        customInterval()
            .delay(2, TimeUnit.SECONDS)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe({ t -> subject.onNext(t) }
            ) { t -> Log.d(RxJava2Learn1.TAG, "Rx customIntervalSub Error $t") }
    }
}